/*
 *  Copyright 1999-2019 Seata.io Group.
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */
package io.seata.rm.datasource.undo.mysql;

import java.util.List;
import java.util.stream.Collectors;

import io.seata.common.exception.ShouldNeverHappenException;
import io.seata.common.util.CollectionUtils;
import io.seata.rm.datasource.ColumnUtils;
import io.seata.rm.datasource.SqlGenerateUtils;
import io.seata.rm.datasource.sql.struct.Field;
import io.seata.rm.datasource.sql.struct.Row;
import io.seata.rm.datasource.sql.struct.TableRecords;
import io.seata.rm.datasource.undo.AbstractUndoExecutor;
import io.seata.rm.datasource.undo.SQLUndoLog;
import io.seata.sqlparser.util.JdbcConstants;

/**
 * The type My sql undo update executor.
 *
 * @author sharajava
 */
public class MySQLUndoUpdateExecutor extends AbstractUndoExecutor {

    /**
     * UPDATE a SET x = ?, y = ?, z = ? WHERE pk1 = ? and pk2 = ?
     */
    private static final String UPDATE_SQL_TEMPLATE = "UPDATE %s SET %s WHERE %s ";

    /**
     * 针对Update操作的undo操作
     */
    @Override
    protected String buildUndoSQL() {
        /** 获得前镜像 */
        TableRecords beforeImage = sqlUndoLog.getBeforeImage();

        /** 获得前镜像的所有行，如果获取不到，则表明update语句没有更新任何行，抛出异常 */
        List<Row> beforeImageRows = beforeImage.getRows();
        if (CollectionUtils.isEmpty(beforeImageRows)) {
            throw new ShouldNeverHappenException("Invalid UNDO LOG");
        }

        /** 获得第1行 */
        Row row = beforeImageRows.get(0);

        /** 获取该行所有的非主键列，拼装需要更新的列 */
        List<Field> nonPkFields = row.nonPrimaryKeys();
        String updateColumns = nonPkFields.stream().
                map(field -> ColumnUtils.addEscape(field.getName(), JdbcConstants.MYSQL) + " = ?").
                collect(Collectors.joining(", "));

        /** 获得主键名，拼装where条件语句 */
        List<String> pkNameList = getOrderedPkList(beforeImage, row, JdbcConstants.MYSQL).stream().
                map(e -> e.getName()).collect(Collectors.toList());
        String whereSql = SqlGenerateUtils.buildWhereConditionByPKs(pkNameList, JdbcConstants.MYSQL);

        /**  构建回滚sql语句，例如：UPDATE [tabelName] SET x = ?, y = ?, z = ? WHERE pk1 = ? and pk2 = ? */
        return String.format(UPDATE_SQL_TEMPLATE, sqlUndoLog.getTableName(), updateColumns, whereSql);
    }

    /**
     * Instantiates a new My sql undo update executor.
     *
     * @param sqlUndoLog the sql undo log
     */
    public MySQLUndoUpdateExecutor(SQLUndoLog sqlUndoLog) {
        super(sqlUndoLog);
    }

    @Override
    protected TableRecords getUndoRows() {
        return sqlUndoLog.getBeforeImage();
    }
}
